package com.navercorp.pinpoint.web.config;

import com.navercorp.pinpoint.common.hbase.*;
import com.navercorp.pinpoint.common.hbase.distributor.RangeOneByteSimpleHash;
import com.navercorp.pinpoint.common.server.util.ThreadPoolExecutorFactoryBean;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.util.ShutdownHookManagerProxy;
import com.sematext.hbase.wd.RowKeyDistributorByHashPrefix;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;

import java.util.Properties;
import java.util.concurrent.ExecutorService;

@Configuration
@ComponentScan({"com.navercorp.pinpoint.common.server.bo", "com.navercorp.pinpoint.common.server.util"})
public class HbaseConfig {

    @Bean
    public HbaseConfigurationFactoryBean hbaseConfiguration() {
        HbaseConfigurationFactoryBean hbaseConfigurationFactoryBean = new HbaseConfigurationFactoryBean();
        Properties properties = new Properties();
        properties.put("hbase.zookeeper.quorum","guns");
        properties.put("hbase.zookeeper.property.clientPort","2181");
        properties.put("zookeeper.znode.parent","/hbase");
        properties.put("hbase.ipc.client.tcpnodelay","true");
        properties.put("hbase.rpc.timeout","60000");
        properties.put("hbase.client.operation.timeout","10000");
        properties.put("hbase.ipc.client.socket.timeout.read","20000");
        properties.put("hbase.ipc.client.socket.timeout.write","20000");
        //properties.put("hbase.rootdir","hdfs://guns:9000/hbase");
        hbaseConfigurationFactoryBean.setProperties(properties);
        return hbaseConfigurationFactoryBean;
    }

    @Bean
    public ThreadPoolExecutorFactoryBean hbaseThreadPool() {
        ThreadPoolExecutorFactoryBean hbaseThreadPool = new ThreadPoolExecutorFactoryBean();
        hbaseThreadPool.setCorePoolSize(64);
        hbaseThreadPool.setMaxPoolSize(64);
        hbaseThreadPool.setQueueCapacity(5120);
        hbaseThreadPool.setThreadNamePrefix("Pinpoint-HConnectionExecutor-");
        hbaseThreadPool.setDaemon(true);
        hbaseThreadPool.setWaitForTasksToCompleteOnShutdown(true);
        hbaseThreadPool.setAwaitTerminationSeconds(10);
        hbaseThreadPool.setPreStartAllCoreThreads(true);
        return hbaseThreadPool;
    }

    @Bean
    public ConnectionFactoryBean connectionFactory(org.apache.hadoop.conf.Configuration hbaseConfiguration, ExecutorService hbaseThreadPool) {
        return new ConnectionFactoryBean(hbaseConfiguration,hbaseThreadPool);
    }

    @Bean
    public HbaseTableFactory hbaseTableFactory(Connection connectionFactory) {
        return new HbaseTableFactory(connectionFactory);
    }

    @Bean
    public HbaseTemplate2 hbaseTemplate(org.apache.hadoop.conf.Configuration hbaseConfiguration,HbaseTableFactory hbaseTableFactory) {
        HbaseTemplate2 hbaseTemplate = new HbaseTemplate2();
        hbaseTemplate.setConfiguration(hbaseConfiguration);
        hbaseTemplate.setTableFactory(hbaseTableFactory);
        hbaseTemplate.setEnableParallelScan(true);
        hbaseTemplate.setMaxThreads(64);
        hbaseTemplate.setMaxThreadsPerParallelScan(16);
        return hbaseTemplate;
    }

    @Bean
    public HbaseAdminFactory hbaseAdminFactory(Connection connectionFactory) {
        return new HbaseAdminFactory(connectionFactory);
    }

    @Bean
    public ShutdownHookManagerProxy shutdownHookManagerProxy() {
        return new ShutdownHookManagerProxy();
    }

    @Bean
    public RowKeyDistributorByHashPrefix.OneByteSimpleHash applicationTraceIndex() {
        return new RowKeyDistributorByHashPrefix.OneByteSimpleHash(32);
    }

    @Bean
    public RowKeyDistributorByHashPrefix applicationTraceIndexDistributor(RowKeyDistributorByHashPrefix.OneByteSimpleHash applicationTraceIndex) {
        return new RowKeyDistributorByHashPrefix(applicationTraceIndex);
    }

    @Bean
    public RangeOneByteSimpleHash traceV2Hasher() {
        return new RangeOneByteSimpleHash(32,40,256);
    }

    @Bean
    public RowKeyDistributorByHashPrefix traceV2Distributor(RangeOneByteSimpleHash traceV2Hasher) {
        return new RowKeyDistributorByHashPrefix(traceV2Hasher);
    }

    @Bean
    public RangeOneByteSimpleHash agentStatV2RangeHasher() {
        return new RangeOneByteSimpleHash(0,33,64);
    }

    @Bean
    public RowKeyDistributorByHashPrefix agentStatV2RowKeyDistributor(RangeOneByteSimpleHash agentStatV2RangeHasher) {
        return new RowKeyDistributorByHashPrefix(agentStatV2RangeHasher);
    }

    @Bean
    public RangeOneByteSimpleHash applicationStatRangeHasher() {
        return new RangeOneByteSimpleHash(0,33,64);
    }

    @Bean
    public RowKeyDistributorByHashPrefix applicationStatRowKeyDistributor(RangeOneByteSimpleHash applicationStatRangeHasher) {
        return new RowKeyDistributorByHashPrefix(applicationStatRangeHasher);
    }

    @Bean
    public RangeOneByteSimpleHash metadataRangeHasher() {
        return new RangeOneByteSimpleHash(0,32,8);
    }

    @Bean
    public RowKeyDistributorByHashPrefix metadataRowKeyDistributor(RangeOneByteSimpleHash metadataRangeHasher) {
        return new RowKeyDistributorByHashPrefix(metadataRangeHasher);
    }

    @Bean
    public RangeOneByteSimpleHash metadataRangeHasher2() {
        return new RangeOneByteSimpleHash(0,36,32);
    }

    @Bean
    public RowKeyDistributorByHashPrefix metadataRowKeyDistributor2(RangeOneByteSimpleHash metadataRangeHasher2) {
        return new RowKeyDistributorByHashPrefix(metadataRangeHasher2);
    }

    @Bean
    public RangeOneByteSimpleHash acceptApplicationHasher() {
        return new RangeOneByteSimpleHash(0,24,4);
    }

    @Bean
    public RowKeyDistributorByHashPrefix acceptApplicationRowKeyDistributor(RangeOneByteSimpleHash acceptApplicationHasher) {
        return new RowKeyDistributorByHashPrefix(acceptApplicationHasher);
    }

    @Bean
    public RangeOneByteSimpleHash statisticsCalleeHasher() {
        return new RangeOneByteSimpleHash(0,36,32);
    }

    @Bean
    public RowKeyDistributorByHashPrefix statisticsCalleeRowKeyDistributor(RangeOneByteSimpleHash statisticsCalleeHasher) {
        return new RowKeyDistributorByHashPrefix(statisticsCalleeHasher);
    }

    @Bean
    public RangeOneByteSimpleHash statisticsCallerHasher() {
        return new RangeOneByteSimpleHash(0,36,32);
    }

    @Bean
    public RowKeyDistributorByHashPrefix statisticsCallerRowKeyDistributor(RangeOneByteSimpleHash statisticsCallerHasher) {
        return new RowKeyDistributorByHashPrefix(statisticsCallerHasher);
    }

    @Bean
    public RangeOneByteSimpleHash statisticsSelfHasher() {
        return new RangeOneByteSimpleHash(0,32,8);
    }

    @Bean
    public RowKeyDistributorByHashPrefix statisticsSelfRowKeyDistributor(RangeOneByteSimpleHash statisticsSelfHasher) {
        return new RowKeyDistributorByHashPrefix(statisticsSelfHasher);
    }

    @Bean
    public HbaseTableNameProvider hbaseTableNameProvider() {
        return new HbaseTableNameProvider("default");
    }


}

